#include "epolllistenserver.h"

#include <sys/time.h>
#include <sys/resource.h>

using namespace std;

void EventFdTimeOutThread::run()
{
    while(!m_exitflag)
    {
        /// 主动关闭超时连接
        set<int>   timeoutFds;
        m_thread.getTimeoutFds(timeoutFds);
        
        int closed = 0;
        for(set<int>::iterator it = timeoutFds.begin();
                it != timeoutFds.end(); ++it)
        {
            if (*it < 0)
                continue;
            
            ::close(*it);
            ++closed;
        }
        
        m_thread.reduceMaxEvent(closed);
        if (closed == 0)
        {
            m_event.wait(1000);
        }
    }
}

EpollEventThread::EpollEventThread(int epollfd,int poolsize,
        unsigned int dispatcher_fdtimeout,
        EpollClientFactory *pf):
    m_epollfd(epollfd)
    ,m_poolsize(poolsize)
    ,m_fdsize(0)
    ,m_exitflag(false)
    ,m_factory(pf)
    ,m_pool(NULL)
    ,m_dispatcher(NULL)
{
    if ( m_poolsize > 0 )
    {
        m_pool = new ThreadPool(0,m_poolsize);
        m_dispatcher = new EpollDispatcher(*m_pool,*m_factory,dispatcher_fdtimeout);
        m_dispatcher->setEpollEventThread(this);
    }
    m_timeoutTarget = new EventFdTimeOutThread(*this);
    m_timeoutThread.start(*m_timeoutTarget);
}

EpollEventThread::~EpollEventThread()
{
    /// 资源类的都需要手动关闭
    if ( m_timeoutTarget != NULL )
    {
        m_timeoutTarget->stop();
        delete m_timeoutTarget;
        m_timeoutTarget = NULL;
    }
    m_timeoutThread.join();

    if ( m_dispatcher != NULL )
    {
        m_dispatcher->stop();
        delete m_dispatcher;
        m_dispatcher = NULL;
    }
    if ( m_pool != NULL )
    {
        m_pool->stopAll();
        delete m_pool;
        m_pool = NULL;
    }
}

void EpollEventThread::closingFd(int timeoutFd)
{
    ScopedMutex   lock(&m_mutex);
    m_timeoutFds.insert(timeoutFd);
    m_event.set();
}

void EpollEventThread::getTimeoutFds(set<int>& timeoutFds)
{
    ScopedMutex   lock(&m_mutex);
    timeoutFds = m_timeoutFds;
    m_timeoutFds.clear();
}

void EpollEventThread::run()
{
    if ( m_dispatcher == NULL )
        return;

    int nfds                    = 0;
    int etimeout                = 2;   //mili second
    
    std::vector<epoll_event>    events;
    unsigned int defaultsize    = 1024;
    events.resize(defaultsize);

    while(!m_exitflag)
    {
        nfds = epoll_wait(m_epollfd,&events[0],events.size(),etimeout);
        for(int i = 0; i < nfds; ++i)
        {
            // 客户端 close 会同时触发 EPOLLIN 和 EPOLLRDHUP，用EPOLLIN只表示可读
            if (events[i].events&EPOLLRDHUP )
            {
                m_dispatcher->dequeue(events[i].data.fd);
                closingFd(events[i].data.fd);
            }
            else if( events[i].events&EPOLLIN )
            {
                m_dispatcher->enqueue(events[i].data.fd);
            }
            // events[i].events&EPOLLERR
            // events[i].events&EPOLLHUP
            else
            {
                m_dispatcher->dequeue(events[i].data.fd);
                closingFd(events[i].data.fd);
            }
        }
        
        ScopedMutex lock(&m_mutex);
        if (events.size() < defaultsize)
            events.resize(defaultsize);
        else if ( events.size() < m_fdsize )
            events.resize( m_fdsize + 10000);
        else if (events.size() - m_fdsize > 10000)
            events.resize(events.size() - 10000);
    }
}

EpollListenServer::EpollListenServer(const SocketAddress& address, 
            EpollClientFactory* pf,
            unsigned int eventepollnum,
            unsigned int dispatcher_threadpoolsize,
            unsigned int dispatcher_fdtimeout,
            bool reuseAddress,
            EPOLL_MODE mode,
            int backlog)
    :m_eventepollnum(eventepollnum)
    ,m_dispatcher_threadpoolsize(dispatcher_threadpoolsize)
    ,m_dispatcher_fdtimeout(dispatcher_fdtimeout)
    ,m_mode(mode)
    ,m_eventepollfds(NULL)
    ,m_factory(pf)
{
    Socket::bind(address, reuseAddress);
    Socket::listen(backlog);
    m_epollFd   = 0;
    m_exitflag  = false;
    init();
}

EpollListenServer::EpollListenServer(unsigned int port,
            EpollClientFactory* pf,
            unsigned int eventepollnum,
            unsigned int dispatcher_threadpoolsize,
            unsigned int dispatcher_fdtimeout,
            bool reuseAddress,
            EPOLL_MODE mode,
            int backlog)
    :m_eventepollnum(eventepollnum)
    ,m_dispatcher_threadpoolsize(dispatcher_threadpoolsize)
    ,m_dispatcher_fdtimeout(dispatcher_fdtimeout)
    ,m_mode(mode)
    ,m_eventepollfds(NULL)
    ,m_factory(pf)
{
    SocketAddress address("0.0.0.0", port);
    Socket::bind(address, reuseAddress);
    Socket::listen(backlog);
    m_epollFd = 0;
    m_exitflag =false;
    init();
}

EpollListenServer& EpollListenServer::operator = (const EpollListenServer& server)
{
    Socket::operator = (server);
    return *this;
}

EpollListenServer::~EpollListenServer()
{
    /// 资源类的都需要手动关闭
    stopWait();

    vector<EpollEventThread*>::iterator itt;
    unsigned int i = 0;
    for( itt = m_threadtargets.begin(); itt != m_threadtargets.end(); ++itt)
    {
        (*itt)->stop();
        m_threads[i]->join();
        delete *itt;
        delete m_threads[i];
        ++i;
    }

    m_threads.clear();
    m_threadtargets.clear();
    for (i = 0; i < m_eventepollnum; ++i )
    {
        ::close(*(m_eventepollfds+i));
    }
    delete [] m_eventepollfds;
    m_eventepollfds = NULL;

    ::close(m_epollFd);
}

void EpollListenServer::init(unsigned long maxopenfd)
{
    struct rlimit rt;
    /* 设置每个进程允许打开的最大文件数 */
    rt.rlim_max = rt.rlim_cur = maxopenfd;
    setrlimit(RLIMIT_NOFILE, &rt);
    
    // epoll init
     /*
     * Since Linux 2.6.8, the size argument is ignored, but must be greater than zero
     */
    m_epollFd   = epoll_create(maxopenfd);
    struct epoll_event ev;
    ev.data.fd  = sockfd();
    if (m_mode == EPOLLMODE_ET)
    {
        ev.events   = EPOLLIN | EPOLLET;
        Socket::setBlocking(sockfd(),false);
    }
    else
    {
        ev.events   = EPOLLIN;
    }
    // 设置监听事件
    epoll_ctl(m_epollFd,EPOLL_CTL_ADD,sockfd(),&ev);
    if ( m_eventepollnum > 0)
    {
        m_eventepollfds = new int[m_eventepollnum];

        for (unsigned int i = 0; i < m_eventepollnum; ++i)
        {
            *(m_eventepollfds + i) = epoll_create(maxopenfd);
            EpollEventThread* ptarget = new EpollEventThread(*(m_eventepollfds + i),
                    m_dispatcher_threadpoolsize,
                    m_dispatcher_fdtimeout,
                    m_factory);
            Thread* pthread = new Thread();
            pthread->start(*ptarget);
            m_threadtargets.push_back(ptarget);
            m_threads.push_back(pthread);
        }
    }
}

void EpollListenServer::wait()
{
    int nfds                    = 0;
    int etimeout                = 2;   //mili second
    
    struct epoll_event          ev;
    ev.data.fd                  = 0;
    if (m_mode == EPOLLMODE_ET)
    {
        ev.events               = EPOLLIN|EPOLLRDHUP|EPOLLET|EPOLLERR;
    }
    else
    {
        ev.events               = EPOLLIN|EPOLLRDHUP|EPOLLERR;
    }
    std::vector<epoll_event>    events;
    int defaultsize = 1024;
    events.resize(defaultsize);

    while(!m_exitflag)
    {
        nfds = epoll_wait(m_epollFd,&events[0],events.size(),etimeout);
        for(int i = 0; i < nfds; ++i)
        {
            if(events[i].data.fd == sockfd())
            {
                int tmpfd = 0;
                while( (tmpfd = acceptConnection())  > 0)
                {
                    ev.data.fd   = tmpfd;
                    if (m_mode == EPOLLMODE_ET)
                    {
                        Socket::setBlocking(tmpfd,false);
                        //Socket::setNoDelay(tmpfd,true);
                    }
                    
                    int index = tmpfd%m_eventepollnum;
                    if( 0 == epoll_ctl(*(m_eventepollfds+index),EPOLL_CTL_ADD,tmpfd,&ev))
                    {
                        m_threadtargets[index]->addMaxEvent(1);
                    }
                }
            }
        }
    }
}

int EpollListenServer::acceptConnection()
{
    return Socket::acceptConnection();
}







